package com.yln.watcher;

import com.yln.watcher.client.ZkClient;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;

import java.util.ArrayList;
import java.util.List;

@Slf4j
@RequiredArgsConstructor
public class WorkersChildrenWatcher implements Watcher {

    private final ZkClient zkClient;

    @Override
    public void process(WatchedEvent event) {
        Event.EventType eventType = event.getType();
        log.info("/workers watcher triggered. eventType: {}", eventType);
        switch (eventType) {
            case NodeChildrenChanged:
                List<String> allZkWorkers = zkClient.findAllWorkers();
                List<String> cacheWorkers = zkClient.getCacheWorkers();
                List<String> zkWorkersComplement = new ArrayList<>(allZkWorkers);
                List<String> cacheWorkersComplement = new ArrayList<>(cacheWorkers);
                // 在allZkWorkers里找出cacheWorkers里不存在的 把他加入cache
                zkWorkersComplement.removeAll(cacheWorkers);
                zkClient.cacheWorkers(zkWorkersComplement);

                // 在cacheWorkers里找出allZkWorkers里不存在的 把他们正在执行的任务扔回/tasks节点下
                // 说明这些worker掉线了，还没完成的任务需要重新分配重新执行
                cacheWorkersComplement.removeAll(allZkWorkers);
                zkClient.rollbackTask(cacheWorkersComplement);
                break;
            default:
                break;
        }
    }
}
